package com.yh.csx.event.store.gateway.event;

import com.google.common.eventbus.Subscribe;
import com.netflix.discovery.converters.Auto;
import com.yh.csx.event.client.thread.NamedThreadFactory;
import com.yh.csx.event.store.gateway.event.dataobject.BizEventDo;
import com.yh.csx.event.store.gateway.event.dataobject.EventTypeDo;
import com.yh.csx.event.store.mapstruct.EventTypeMapStruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.domain.Example;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.yh.csx.event.client.event.EventBusUtils.register;

/**
 * @author zhangsheng
 * @date 2021/8/23 18:03
 */
@Slf4j
@Component
public class EventTypeGateway implements CommandLineRunner {
    private static final Map<String, Map<String, BizEventDo>> APP_TYPE_EVENT = new ConcurrentHashMap<>();

    @Value("${bsf.event.rpc.event.type.merge.seconds:300}")
    private int mergeDelaySeconds = 60 * 5;
    @Autowired
    private EventTypeRepository eventTypeRepository;
    @Autowired
    private EventTypeMapStruct eventTypeMapStruct;

    private ScheduledExecutorService timer;

    public void init() {
        register(this);
        timer = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("EventType", true));
        timer.scheduleWithFixedDelay(() -> mergeType(), 10, mergeDelaySeconds, TimeUnit.SECONDS);
    }

    @Subscribe
    public void receiveType(BizEventDo type) {
        if (null == type.getAppName() || null == type.getEventType()) {
            log.warn("=receiveType=>eventType or appName field is null ,check type:{}", type);
            return;
        }
        try {
            Map<String, BizEventDo> typeEventMap = APP_TYPE_EVENT.computeIfAbsent(type.getAppName(), app -> new ConcurrentHashMap<>());
            typeEventMap.computeIfAbsent(type.getEventType(), eventType -> type);
        } catch (Exception e) {
        }
    }

    private void mergeType() {
        APP_TYPE_EVENT.entrySet().forEach(app -> {
            app.getValue().entrySet().forEach(entry -> {
                log.info("=mergeType=>appName:{} ,eventType:{} ,eventTypeDo:{}", app.getKey(), entry.getKey(), entry.getValue());
                EventTypeDo type = new EventTypeDo();
                type.setAppName(app.getKey());
                type.setEventType(entry.getKey());
                List<EventTypeDo> list = eventTypeRepository.findAll(Example.of(type));
                if (CollectionUtils.isEmpty(list)) {
                    eventTypeRepository.save(eventTypeMapStruct.config(entry.getValue()));
                } else {
                    EventTypeDo config = eventTypeMapStruct.config(entry.getValue());
                    list.get(0).setEventType(config.getEventType());
                    list.get(0).setProjectId(config.getProjectId());
                    list.get(0).setAppName(config.getAppName());
                    list.get(0).setTopic(config.getTopic());
                    list.get(0).setRemark("sys");
                    list.get(0).setUpdateTime(new Date());
                    eventTypeRepository.flush();
                }

            });
        });

    }

    @Override
    public void run(String... args) throws Exception {
        init();
    }

}
